package com.fanxuankai.boot.canal.client;

import cn.hutool.core.thread.ThreadUtil;
import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import com.fanxuankai.boot.canal.client.autoconfigure.Handle;
import com.fanxuankai.commons.util.DingTalkClientHelper;
import com.fanxuankai.commons.util.concurrent.Flow;
import com.fanxuankai.commons.util.concurrent.SubmissionPublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 处理订阅者
 *
 * @author fanxuankai
 */
public class HandleSubscriber extends SubmissionPublisher<MessageWrapper>
        implements Flow.Subscriber<MessageWrapper> {
    private static final Logger LOGGER = LoggerFactory.getLogger(HandleSubscriber.class);
    private final OtterClient otterClient;
    private final CanalClientProperties properties;
    private final MessageConsumer messageConsumer;
    private Flow.Subscription subscription;

    public HandleSubscriber(OtterClient otterClient, CanalClientProperties properties,
                            MessageConsumer messageConsumer,
                            ExecutorService executorService) {
        super(executorService, Flow.defaultBufferSize());
        this.otterClient = otterClient;
        this.properties = properties;
        this.messageConsumer = messageConsumer;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(MessageWrapper item) {
        int retry = 0;
        long start = System.currentTimeMillis();
        Handle handle = properties.getHandle();
        while (true) {
            try {
                messageConsumer.accept(item);
                break;
            } catch (Throwable throwable) {
                if (++retry == handle.getRetry()) {
                    throw new RuntimeException(throwable);
                }
                ThreadUtil.sleep(handle.getIntervalMillis(), TimeUnit.MILLISECONDS);
            }
        }
        if (Objects.equals(properties.isShowEventLog(), Boolean.TRUE)
                && !item.getEntryWrapperList().isEmpty()) {
            LOGGER.info("[" + properties.getId() + "] " + "Handle batchId: {} time: {}ms", item.getBatchId(),
                    System.currentTimeMillis() - start);
        }
        submit(item);
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        LOGGER.error("[" + properties.getId() + "] " + throwable.getMessage(), throwable);
        if (properties.getDingTalk().isEnabled()) {
            DingTalkClientHelper dingTalkClientHelper = DingTalkClientHelper.newInstance(properties.getDingTalk().getUrl(),
                    properties.getDingTalk().getAccessToken(),
                    properties.getDingTalk().getSecret());
            String title = "Canal 客户端处理异常";
            dingTalkClientHelper.sendRobotMarkdown(title, "# " + title + "\n"
                    + "> instance: " + properties.getId() + "\n\n"
                    + "> message: " + throwable.getMessage());
        }
        this.subscription.cancel();
        this.otterClient.stop();
    }

    @Override
    public void onComplete() {
        LOGGER.info("[" + properties.getId() + "] " + "Done");
    }
}
